﻿/******************************************************************************
 * This example                                                               *
 * - emulates a client launching a request every 10-300ms                     *
 * - uses a CURL-backend consisting of a master and 10 workers                *
 * - runs until it is shut down by a CTRL+C signal                            *
 *                                                                            *
 *                                                                            *
 * Schematic view:                                                            *
 *                                                                            *
 *    client      |    client_job    |    curl_master    |    curl_worker     *
 *          /--------------|*|-------------\       /-------------|*|          *
 *         /---------------|*|--------------\     /                           *
 *        /----------------|*|---------------\   /                            *
 *     |*| ----------------|*|----------------|*|----------------|*|          *
 *        \________________|*|_______________/   \                            *
 *         \_______________|*|______________/     \                           *
 *          \______________|*|_____________/       \-------------|*|          *
 *                                                                            *
 *                                                                            *
 * Communication pattern:                                                     *
 *                                                                            *
 *        client_job      curl_master        curl_worker                      *
 *          |                  |                  |                           *
 *          | ----(read)-----> |                  |                           *
 *          |                  | --(forward)----> |                           *
 *          |                                     |---\                       *
 *          |                                     |   |                       *
 *          |                                     |<--/                       *
 *          | <-------------(reply)-------------- |                           *
 *          X                                                                 *
 ******************************************************************************/

// C includes
#include <csignal>
#include <cstdlib>
#include <ctime>

// C++ includes
#include <iostream>
#include <random>
#include <string>
#include <vector>

// CAF
#include "caf/all.hpp"
#include "caf/io/all.hpp"

CAF_PUSH_WARNINGS
#include <curl/curl.h>
CAF_POP_WARNINGS

// disable some clang warnings here caused by CURL macros
#ifdef __clang__
#pragma clang diagnostic ignored "-Wshorten-64-to-32"
#pragma clang diagnostic ignored "-Wdisabled-macro-expansion"
#pragma clang diagnostic ignored "-Wunused-const-variable"
#endif // __clang__

CAF_BEGIN_TYPE_ID_BLOCK(curl_fuse, first_custom_type_id)

CAF_ADD_TYPE_ID(curl_fuse, (std::vector<char>))

CAF_ADD_ATOM(curl_fuse, read_atom)
CAF_ADD_ATOM(curl_fuse, fail_atom)
CAF_ADD_ATOM(curl_fuse, next_atom)
CAF_ADD_ATOM(curl_fuse, reply_atom)
CAF_ADD_ATOM(curl_fuse, finished_atom)

CAF_END_TYPE_ID_BLOCK(curl_fuse)

using namespace caf;

using buffer_type = std::vector<char>;

namespace color
{

    // UNIX terminal color codes
    constexpr char reset[] = "\033[0m";
    constexpr char reset_endl[] = "\033[0m\n";
    constexpr char black[] = "\033[30m";
    constexpr char red[] = "\033[31m";
    constexpr char green[] = "\033[32m";
    constexpr char yellow[] = "\033[33m";
    constexpr char blue[] = "\033[34m";
    constexpr char magenta[] = "\033[35m";
    constexpr char cyan[] = "\033[36m";
    constexpr char white[] = "\033[37m";
    constexpr char bold_black[] = "\033[1m\033[30m";
    constexpr char bold_red[] = "\033[1m\033[31m";
    constexpr char bold_green[] = "\033[1m\033[32m";
    constexpr char bold_yellow[] = "\033[1m\033[33m";
    constexpr char bold_blue[] = "\033[1m\033[34m";
    constexpr char bold_magenta[] = "\033[1m\033[35m";
    constexpr char bold_cyan[] = "\033[1m\033[36m";
    constexpr char bold_white[] = "\033[1m\033[37m";

} // namespace color

// number of HTTP workers
constexpr size_t num_curl_workers = 10;

// minimum delay between HTTP requests
constexpr int min_req_interval = 10;

// maximum delay between HTTP requests
constexpr int max_req_interval = 300;

// put everything into anonymous namespace (except main)
namespace
{

    // provides print utility, a name, and a parent
    struct base_state
    {
        base_state(local_actor *thisptr) : self(thisptr)
        {
            // nop
        }

        actor_ostream print()
        {
            return aout(self) << color << self->name() << " (id = " << self->id()
                              << "): ";
        }

        virtual bool init(std::string m_color)
        {
            color = std::move(m_color);
            print() << "started" << color::reset_endl;
            return true;
        }

        virtual ~base_state()
        {
            print() << "done" << color::reset_endl;
        }

        local_actor *self;
        std::string color;
    };

    struct client_job_state : base_state
    {
        static inline const char *name = "curl.client-job";
        using base_state::base_state;
    };

    // encapsulates an HTTP request
    behavior client_job(stateful_actor<client_job_state> *self,
                        const actor &parent)
    {
        if (!self->state.init(color::blue))
            return {}; // returning an empty behavior terminates the actor
        self->send(parent, read_atom_v, "http://www.example.com/index.html",
                   uint64_t{0}, uint64_t{4095});
        return {
            [=](reply_atom, const buffer_type &buf)
            {
                self->state.print() << "successfully received " << buf.size() << " bytes"
                                    << color::reset_endl;
                self->quit();
            },
            [=](fail_atom)
            {
                self->state.print() << "failure" << color::reset_endl;
                self->quit();
            },
        };
    }

    struct client_state : base_state
    {
        client_state(local_actor *selfptr)
            : base_state(selfptr),
              count(0),
              re(rd()),
              dist(min_req_interval, max_req_interval)
        {
            // nop
        }

        size_t count;
        std::random_device rd;
        std::default_random_engine re;
        std::uniform_int_distribution<int> dist;
        static inline const char *name = "curl.client";
    };

    // spawns HTTP requests
    behavior client(stateful_actor<client_state> *self, const actor &parent)
    {
        using std::chrono::milliseconds;
        self->link_to(parent);
        if (!self->state.init(color::green))
            return {}; // returning an empty behavior terminates the actor
        self->send(self, next_atom_v);
        return {
            [=](next_atom)
            {
                auto &st = self->state;
                st.print() << "spawn new client_job (nr. " << ++st.count << ")"
                           << color::reset_endl;
                // client_job will use IO
                // and should thus be spawned in a separate thread
                self->spawn<detached + linked>(client_job, parent);
                // compute random delay until next job is launched
                auto delay = st.dist(st.re);
                self->delayed_send(self, milliseconds(delay), next_atom_v);
            },
        };
    }

    struct curl_state : base_state
    {
        curl_state(local_actor *selfptr) : base_state(selfptr)
        {
            // nop
        }

        ~curl_state() override
        {
            if (curl != nullptr)
                curl_easy_cleanup(curl);
        }

        static size_t callback(void *data, size_t bsize, size_t nmemb, void *userp)
        {
            size_t size = bsize * nmemb;
            auto &buf = reinterpret_cast<curl_state *>(userp)->buf;
            auto first = reinterpret_cast<char *>(data);
            auto last = first + bsize;
            buf.insert(buf.end(), first, last);
            return size;
        }

        bool init(std::string m_color) override
        {
            curl = curl_easy_init();
            if (curl == nullptr)
                return false;
            curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, &curl_state::callback);
            curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
            return base_state::init(std::move(m_color));
        }

        CURL *curl = nullptr;
        buffer_type buf;
        static inline const char *name = "curl.worker";
    };

    // manages a CURL session
    behavior curl_worker(stateful_actor<curl_state> *self, const actor &parent)
    {
        if (!self->state.init(color::yellow))
            return {}; // returning an empty behavior terminates the actor
        return {[=](read_atom, const std::string &fname, uint64_t offset,
                    uint64_t range) -> message
                {
                    auto &st = self->state;
                    st.print() << "read" << color::reset_endl;
                    for (;;)
                    {
                        st.buf.clear();
                        // set URL
                        curl_easy_setopt(st.curl, CURLOPT_URL, fname.c_str());
                        // set range
                        std::ostringstream oss;
                        oss << offset << "-" << range;
                        curl_easy_setopt(st.curl, CURLOPT_RANGE, oss.str().c_str());
                        // set curl callback
                        curl_easy_setopt(st.curl, CURLOPT_WRITEDATA,
                                         reinterpret_cast<void *>(&st));
                        // launch file transfer
                        auto res = curl_easy_perform(st.curl);
                        if (res != CURLE_OK)
                        {
                            st.print() << "curl_easy_perform() failed: " << curl_easy_strerror(res)
                                       << color::reset_endl;
                        }
                        else
                        {
                            long hc = 0; // http return code
                            curl_easy_getinfo(st.curl, CURLINFO_RESPONSE_CODE, &hc);
                            switch (hc)
                            {
                            default:
                                st.print() << "http error: download failed with "
                                           << "'HTTP RETURN CODE': " << hc << color::reset_endl;
                                break;
                            case 200: // ok
                            case 206: // partial content
                                st.print() << "received " << st.buf.size()
                                           << " bytes with 'HTTP RETURN CODE': " << hc
                                           << color::reset_endl;
                                // tell parent that this worker is done
                                self->send(parent, finished_atom_v);
                                return make_message(reply_atom_v, std::move(st.buf));
                            case 404: // file does not exist
                                st.print() << "http error: download failed with "
                                           << "'HTTP RETURN CODE': 404 (file does "
                                           << "not exist!)" << color::reset_endl;
                            }
                        }
                        // avoid 100% cpu utilization if remote side is not accessible
                        std::this_thread::sleep_for(std::chrono::milliseconds(100));
                    }
                }};
    }

    struct master_state : base_state
    {
        master_state(local_actor *selfptr) : base_state(selfptr)
        {
            // nop
        }
        std::vector<actor> idle;
        std::vector<actor> busy;
        static inline const char *name = "curl.master";
    };

    behavior curl_master(stateful_actor<master_state> *self)
    {
        if (!self->state.init(color::magenta))
            return {}; // returning an empty behavior terminates the actor
        // spawn workers
        for (size_t i = 0; i < num_curl_workers; ++i)
            self->state.idle.push_back(
                self->spawn<detached + linked>(curl_worker, self));
        auto worker_finished = [=]
        {
            auto sender = self->current_sender();
            auto last = self->state.busy.end();
            auto i = std::find(self->state.busy.begin(), last, sender);
            if (i == last)
                return;
            self->state.idle.push_back(*i);
            self->state.busy.erase(i);
            self->state.print() << "worker is done" << color::reset_endl;
        };
        self->state.print() << "spawned " << self->state.idle.size() << " worker(s)"
                            << color::reset_endl;
        return {
            [=](read_atom rd, std::string &str, uint64_t x, uint64_t y)
            {
                auto &st = self->state;
                st.print() << "received {'read'}" << color::reset_endl;
                // forward job to an idle worker
                actor worker = st.idle.back();
                st.idle.pop_back();
                st.busy.push_back(worker);
                self->delegate(worker, rd, std::move(str), x, y);
                st.print() << st.busy.size() << " active jobs" << color::reset_endl;
                if (st.idle.empty())
                {
                    // wait until at least one worker finished its job
                    self->become(keep_behavior, [=](finished_atom)
                                 {
          worker_finished();
          self->unbecome(); });
                }
            },
            [=](finished_atom)
            { worker_finished(); },
        };
    }

    // signal handling for ctrl+c
    std::atomic<bool> shutdown_flag{false};

} // namespace

void caf_main(actor_system &system)
{
    // install signal handler
    struct sigaction act;
    act.sa_handler = [](int)
    { shutdown_flag = true; };
    auto set_sighandler = [&]
    {
        if (sigaction(SIGINT, &act, nullptr) != 0)
        {
            std::cerr << "fatal: cannot set signal handler" << std::endl;
            abort();
        }
    };
    set_sighandler();
    // initialize CURL
    curl_global_init(CURL_GLOBAL_DEFAULT);
    // get a scoped actor for the communication with our CURL actors
    scoped_actor self{system};
    // spawn client and curl_master
    auto master = self->spawn<detached>(curl_master);
    self->spawn<detached>(client, master);
    // poll CTRL+C flag every second
    while (!shutdown_flag)
        std::this_thread::sleep_for(std::chrono::seconds(1));
    aout(self) << color::cyan << "received CTRL+C" << color::reset_endl;
    // shutdown actors
    anon_send_exit(master, exit_reason::user_shutdown);
    // await actors
    act.sa_handler = [](int)
    { abort(); };
    set_sighandler();
    aout(self) << color::cyan
               << "await CURL; this may take a while "
                  "(press CTRL+C again to abort)"
               << color::reset_endl;
    self->await_all_other_actors_done();
    // shutdown CURL
    curl_global_cleanup();
}

CAF_MAIN(id_block::curl_fuse, io::middleman)